#include <moying_comm_lib/reply.h>
#include <moying_comm_lib/utility.h>
#include <thread>

namespace moying
{
namespace communication
{

MoyingCommuReply::MoyingCommuReply(const std::string &url,
                             const std::string &identity) : router_(MoyingCommuBase::ctx_, zmq::socket_type::router),
                                                            dealer_(MoyingCommuBase::ctx_, zmq::socket_type::dealer),
                                                            ctrl_pub_(MoyingCommuBase::ctx_, zmq::socket_type::pub),
                                                            ctrl_sub_(MoyingCommuBase::ctx_, zmq::socket_type::sub)
{
  MoyingCommuBase::url_ = url;
  MoyingCommuBase::identity_ = identity;
  recv_handler_thread_number_ = 1;
}

MoyingCommuReply::MoyingCommuReply(const std::string &ip, const unsigned int port, const std::string &identity)
:MoyingCommuReply(std::string("tcp://" + ip + ":" + std::to_string(port)), identity)
{
}

MoyingCommuReply::~MoyingCommuReply()
{
  close();
}

bool MoyingCommuReply::run()
{

  if (is_started_)
    return false;

  try
  {
    if(heartbeat_timeout_ms_ > 0){
      router_.setsockopt(ZMQ_HEARTBEAT_IVL, heartbeat_interval_ms_);
      router_.setsockopt(ZMQ_HEARTBEAT_TIMEOUT, heartbeat_timeout_ms_);
    }
    router_.setsockopt(ZMQ_LINGER, kLingerTimeout);
    if(heartbeat_timeout_ms_ > 0){
      dealer_.setsockopt(ZMQ_HEARTBEAT_IVL, heartbeat_interval_ms_);
      dealer_.setsockopt(ZMQ_HEARTBEAT_TIMEOUT, heartbeat_timeout_ms_);
    }
    dealer_.setsockopt(ZMQ_LINGER, kLingerTimeout);

    std::ostringstream o;
    o<<serial_++;
    std::string inproc_name("inproc://monitor-router");
    inproc_name += o.str();

    monitor_start_waiter_->set_signaled(false);
    createMonitor(router_, inproc_name);
    if(!monitor_start_waiter_->signaled())  //防止notify后再wait
        monitor_start_waiter_->wait();

    router_.bind(url_);
    std::ostringstream o2;
    o2<<serial_++;
    worker_inproc_name_ = std::string("inproc://workers");
    worker_inproc_name_ += o2.str();
    dealer_.bind(worker_inproc_name_);

    std::ostringstream o3;
    o3<<serial_++;
    std::string ctrl_inproc_name_ = std::string("inproc://ctrl");
    ctrl_inproc_name_ += o3.str();
    ctrl_pub_.bind(ctrl_inproc_name_);
    ctrl_sub_.connect(ctrl_inproc_name_);
    ctrl_sub_.setsockopt(ZMQ_SUBSCRIBE,"",0);

    createWorker();
    createProxy();
  }
  catch (zmq::error_t e)
  {
    callLog(MoyingCommuLogLevels::FATAL, "MoyingCommuReply::run() throw %s err_code:%d\n",
            e.what(), e.num());
    return false;
  }
  is_started_ = true;
  return true;
}

bool MoyingCommuReply::close()
{

  if(is_stoped_)
      return false;

  callLog(MoyingCommuLogLevels::TRACE, "%s","enter reply mode socket close\n");
  //关闭proxy
  std::string subscriber =  "TERMINATE";
  zmq::message_t msg_content(subscriber.data(), subscriber.size());
  ctrl_pub_.send(msg_content);

  is_stoped_ = true;
  while (is_started_)
  {
    SLEEP(10);
    if (is_loop_exit_ && is_monitor_exit_)
      break;
  }
  is_started_ = false;
  router_.close();
  dealer_.close();

  callLog(MoyingCommuLogLevels::TRACE, "%s","leave reply mode socket close\n");
  return true;
}

/**
 * @brief printPackWrapper      打印通讯数据包
 * @param is_send               true:发送; false:接收
 * @param pack                  数据包
 * @return
 */
void MoyingCommuReply::printPackWrapper(bool is_send, const std::string &content, bytes_ptr pack, int thread_id)
{
  if (print_pack_call)
  {
    if(proto_types_.find(content) == proto_types_.end())
    {
      WARNING("Content "<<content<<" not found in registered proto types");
      return;
    }

    char* p = pack->data();
    int total_size  = GET_INT(p);
    int header_size = GET_INT(p);
    p = pack->data();
    bytes_ptr data = std::make_shared<bytes_vec>(p+header_size , p+total_size);
    OFFSET(p,8);
    int identity_len = GET_INT(p);
    std::string identity(p,identity_len);
    OFFSET(p,identity_len);
    int pack_id = GET_INT(p);
    long long timestamp = GET_LONGLONG(p);
    timestamp = ntohll(timestamp);

    moying_proto::ServicePack whole_pack;
    whole_pack.ParseFromArray(data->data(),data->size());
    std::string data_str;
    if(is_send)
    {
      proto_types_[content].second->ParseFromArray(whole_pack.response().data(),whole_pack.response().size());
      google::protobuf::TextFormat::PrintToString(*proto_types_[content].second, &data_str);
    }
    else
    {
      proto_types_[content].first->ParseFromArray(whole_pack.request().data(),whole_pack.request().size());
      google::protobuf::TextFormat::PrintToString(*proto_types_[content].first, &data_str);
    }
    
    // std::string fancy_msg = "\n\033[1m\033[30m[Moying]:\033[0m" + std::to_string(timestamp) 
    //             + (is_send ? " [Request: " : " [Response: " ) + identity + "-\033[35m" + content + "\033[0m-" + std::to_string(pack_id) + "]\n";
    std::string time_str = getDateTime(timestamp);
    std::string fancy_msg = "\n[Moying]:" + time_str 
                + (is_send ? " [Response: " : " [Request: " ) + identity + "-" + content + "-" + std::to_string(pack_id) + "]\n";
    print_pack_call(fancy_msg + data_str);
  }
}

/**
 * @brief createProxy           创建监听代理转发线程
 * @return
 */
void MoyingCommuReply::createProxy()
{
  auto pkg = [&]() -> void {
    auto thread_id = getTid();
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] started proxy.\n", thread_id);
    try
    {
      zmq::proxy_steerable(router_, dealer_, nullptr,ctrl_sub_);
    }
    catch (zmq::error_t e)
    {
      callLog(MoyingCommuLogLevels::FATAL, "[tid:%d] MoyingCommuReply proxy throw %s err_code:%d\n",
              thread_id, e.what(), e.num());
    }
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] stoped proxy.\n", thread_id);
    is_loop_exit_ = true;
  };

  std::thread thr(pkg);
  thr.detach();
}

/**
 * @brief createWorker          创建请求处理线程
 * @return
 */
void MoyingCommuReply::createWorker()
{
  auto pkg = [&]() -> void {
    auto thread_id = getTid();
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] started worker.\n", thread_id);

    std::vector<zmq_pollitem_t> poll_vec;
    std::unique_ptr<zmq::socket_t> worker;
    try
    {
      worker.reset(new zmq::socket_t(ctx_, zmq::socket_type::rep));
      worker->setsockopt(ZMQ_HEARTBEAT_IVL, heartbeat_interval_ms_);
      worker->setsockopt(ZMQ_HEARTBEAT_TIMEOUT, heartbeat_timeout_ms_);
      worker->setsockopt(ZMQ_LINGER, kLingerTimeout);
      worker->connect(worker_inproc_name_);
      poll_vec.push_back({*worker, 0, ZMQ_POLLIN, 0});
    }
    catch (zmq::error_t e)
    {
      callLog(MoyingCommuLogLevels::FATAL, "[tid:%d] MoyingCommuRequest worker init throw %s err_code:%d\n",
              thread_id, e.what(), e.num());
    }

    while (!is_stoped_)
    {
      try
      {
        bytes_ptr pack_recv = nullptr;
        bytes_ptr reply_data = nullptr;
        std::string content;
        bool print_pack = false;

        // 拉取请求数据
        zmq::poll(poll_vec, poll_timeout_ms_); // 这里超时并不影响业务, 与request模式不同
        if (poll_vec[0].revents & ZMQ_POLLIN)
        {

          zmq::message_t msg_recv;
          bool bl = worker->recv(&msg_recv);
          // 解析收取的数据包, 根据需要打印内容
          pack_recv = std::make_shared<bytes_vec>((char*)msg_recv.data(),(char*)msg_recv.data()+msg_recv.size());

          // 调用接收回调函数
          bytes_ptr req, res;
          decodeServicePack(decodeRecvBuf(pack_recv), content, req, res);
          if(rep_cbs_.find(content) != rep_cbs_.end())
          {
            rep_cbs_[content].first->processService(req, res);
            encodeServicePack(content, req, res, reply_data);
            if(rep_cbs_[content].second){
              print_pack = true;
              printPackWrapper(false, content, pack_recv, thread_id);
            }
          }else{
            encodeServicePack(content, req, res, reply_data);
            // printPackWrapper(false, content, pack_recv, thread_id);
          }

        }
        else
        {
          continue;
        }

        if(pack_recv == nullptr){
          callLog(MoyingCommuLogLevels::FATAL, "[tid:%d] recv pack failed\n",
                  thread_id);
          break;
        }

        //获取发送包的pack_id
        char* p = pack_recv->data();
        OFFSET(p,8);
        int identity_len = GET_INT(p);
        OFFSET(p,identity_len);
        int pack_id = GET_INT(p);

        // 构造完整数据包
        auto pack_send = encodeSendBuf(reply_data, pack_id);

        // 发送 protobuf 数据, 根据需要打印内容
        zmq::message_t msg_send(pack_send->data(), pack_send->size());
        worker->send(msg_send, ZMQ_NOBLOCK);
        if(print_pack)
          printPackWrapper(true, content, pack_send, thread_id);
      }
      catch (zmq::error_t e)
      {
        callLog(MoyingCommuLogLevels::FATAL, "[tid:%d] MoyingCommuRequest worker throw %s err_code:%d\n",
                thread_id, e.what(), e.num());
        if (e.num() == ETERM) // 上下文已终止
          break;
      }
    }
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] stoped worker.\n", thread_id);
  };

  int thr_num = recv_handler_thread_number_;
  if(thr_num <= 0)
    thr_num = 1;
  else if(thr_num > 30)
    thr_num = 30;
  for (int i = 0; i < thr_num; ++i)
  {
    std::thread thr(pkg);
    thr.detach();
  }
}

bool MoyingCommuReply::alterRepContent(const std::string &content, bool add)
{
  if(add && rep_cbs_.find(content) != rep_cbs_.end())
  {
    printf("Invalid service. Confilicting subscription on topic %s", content.c_str());
    return false;
  }
  else if(!add && rep_cbs_.find(content) == rep_cbs_.end())
  {
    printf("Invalid service. Unregistered content topic %s", content.c_str());
    return false;
  }
  if (content.size() == 0)
  {
    printf("Empty content");
    return false;
  }
  return true;
}

void MoyingCommuReply::encodeServicePack(const std::string &content, const bytes_ptr &request, const bytes_ptr &response, bytes_ptr &pack_data)
{
  moying_proto::ServicePack pack;
  pack.set_topic(content);
  pack.set_request(std::string(request->begin(), request->end()));
  pack.set_response(std::string(response->begin(), response->end()));
  std::string msg_str;
  pack.SerializeToString(&msg_str);
  pack_data = std::make_shared<bytes_vec>(msg_str.data(), msg_str.data() + msg_str.length());
}

void MoyingCommuReply::decodeServicePack(const bytes_ptr &pack_data, std::string &content, bytes_ptr &request, bytes_ptr &response)
{
  moying_proto::ServicePack pack;
  // ::google::protobuf::Message *msg = &pack;
  // if(!msg->ParseFromArray(pack_data->data(),pack_data->size()))
  if(!pack.ParseFromArray(pack_data->data(),pack_data->size()))
  {
      printf("%s\n","sub_recv_func parse protobuf failed in decodeRequestPack()");
  }
  content = pack.topic();
  request = std::make_shared<bytes_vec>(pack.request().data(), pack.request().data() + pack.request().length());
  response = std::make_shared<bytes_vec>(pack.response().data(), pack.response().data() + pack.response().length());
}

} // namepsace communication
} // namespace moying
